package com.ferry.lock.engine.zookeeper.menagerie;

import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * Created by daiyong
 */
public class ZKCommandExecutor {
	private static final Logger logger = LoggerFactory.getLogger(ZKCommandExecutor.class);

	private static final long DEFAULT_DELAY_MILLIS = 1000;
	public static final int DEFAULT_MAX_RETRIES = 3;

	private final long delayMillis;
	private final int maxRetries;
	private final ZKSessionManager sessionManager;


	public ZKCommandExecutor(ZKSessionManager sessionManager) {
		this(sessionManager,DEFAULT_MAX_RETRIES,DEFAULT_DELAY_MILLIS, TimeUnit.MILLISECONDS);
	}

	public ZKCommandExecutor(ZKSessionManager sessionManager,long retryDelay,TimeUnit retryUnits) {
		this(sessionManager,DEFAULT_MAX_RETRIES,retryDelay,retryUnits);
	}

	public ZKCommandExecutor(ZKSessionManager sessionManager,int maxRetries) {
		this(sessionManager,maxRetries,DEFAULT_DELAY_MILLIS,TimeUnit.MILLISECONDS);
	}

	public ZKCommandExecutor(ZKSessionManager sessionManager, int maxRetries,long delayMillis, TimeUnit retryUnit){
		this.sessionManager = sessionManager;
		this.maxRetries = maxRetries;
		this.delayMillis = retryUnit.toMillis(delayMillis);
	}

	public <T> T execute(ZKCommand<T> command) throws KeeperException {

		KeeperException exception = null;

		for(int retry=0; retry<maxRetries; retry++){
			try {
				return command.execute(sessionManager.getZooKeeper());
			} catch (InterruptedException e) {
				logger.error("command 被打断", e);
				doWait(retry, delayMillis);
			} catch (KeeperException.ConnectionLossException kce) {
				logger.error("command 执行期间连接断开", kce);
				exception = exception == null ? kce : exception;
				doWait(retry, delayMillis);
			} catch (KeeperException.SessionExpiredException kse) {
				logger.error("command 执行期间会话断开", kse);
				exception = exception == null ? kse : exception;
				throw kse;
			}
		}

		if(exception==null){
			throw new RuntimeException("Repeatedly interrupted during an uninterruptible task caused Task to fail");
		}
		throw exception;
	}

	public <T> T executeInterruptibly(ZKCommand<T> command) throws InterruptedException, KeeperException {
		KeeperException exception = null;
		for(int retry=0;retry<maxRetries;retry++){
			try{
				return command.execute(sessionManager.getZooKeeper());
			} catch (KeeperException.ConnectionLossException kce) {
				if(exception==null){
					exception = kce;
				}
				logger.debug("Connection was lost while comamnd "+ command +"was attempted. Retrying");
				doWaitInterruptibly(retry, delayMillis);
			} catch(KeeperException.SessionExpiredException kse){
				if(exception==null){
					exception = kse;
				}
				logger.debug("Session expired while comamnd "+ command +"was attempted. Retrying with new session");
				doWaitInterruptibly(retry, delayMillis);
			}
		}
		throw exception;
	}

	/**
	 * 等待
	 * @param retry
	 * @param delayMillis
	 */
	private void doWait(int retry, long delayMillis) {
		try {
			Thread.sleep(retry * delayMillis);
		} catch (InterruptedException e) {
			//ignore
		}
	}

	private void doWaitInterruptibly(int retry, long retryDelay) throws InterruptedException{
		Thread.sleep(retry * retryDelay);
	}

}
